/*
 * NIOObjectContainer.java    0.0.1    26/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */ 
package org.mentalsmash.tazio.net.cs.protocol;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.io.StreamCorruptedException;
import java.nio.ByteBuffer;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class serves as a mean to transport over the network serializable objects,
 * using the Non-Blocking Server Architecture as transport. 
 * <p>
 * In fact the {@link org.mentalsmash.tazio.net.cs.server.ProtocolHandler ProtocolHandler}
 * interface returns incoming messages as ByteBuffers, and needs a mechanism to recognise
 * when a complete serializable object has been received, to appropriately pass it to its
 * handler.
 * </p>
 * <p>
 * Therefore the <tt>NIOObjectContainer</tt> acts as an holder for a serializable object
 * and serializes itself in a predictable way using the methods {@link #writeContainer()}
 * and {@link #readContainer(byte[])}
 * </p>
 * <p>
 * What it does is to use the default java serialization mechanisms to serialize the objects
 * it wraps (called the <code>payload</code>) into a byte array.
 * Then it builds a message made up of this parts:
 * <ol>
 *   <li>4 bytes as a predifined marker having the value {@link MessageHeader#START_MARKER}</li>
 *   <li>4 bytes containing the length of the byte array in which the <code>payload</code> is
 *       encoded</li>
 *   <li>The byte array containing the payload itself</li>
 * </ol>
 * Thus, when the ProtocolHandler receives new input from the dispatcher it can easely detect
 * the boundaries of a complete objects, unmarshal the <tt>NIOObjectContainer</tt> and unwrap
 * its contents into a deserialized object.
 * </p>
 * @version    0.0.1    26/set/2009    
 * @author     Andrea Reale <andrea.reale@gmail.com>
 */
public final class NIOObjectContainer {

	
	private final static transient Logger log = LoggerFactory.getLogger(NIOObjectContainer.class);
	
	/**
	 * A 4 bytes constant indicating a null <code>payload</code> in the externalized representation
	 * of <tt>NIOObjectContainer</tt><br />
	 * It Currently is: <code>0xDEADCAFE</code>
	 */
	public final static  int EMPTY_PAYLOAD = 0xDEADCAFE;
	
	private Serializable payload = null;
	

	
	/**
	 * Gets the payload for this ObjectContainer
	 * @return the payload
	 */
	public Serializable getPayload() {
		return payload;
	}

	/**
	 * Sets the payload for this objectContariner
	 * @param payload the payload to set
	 */
	public void setPayload(Serializable payload) {
		this.payload = payload;
	}


	/**
	 * Empty constructor. Sets the <code>payload</code> as <code>null</code>;
	 */
	public NIOObjectContainer() {
		
	}
	
	/**
	 * Create a new <tt>NIOObjectContainer</tt> and sets the given <code>payload</code>
	 * as the wrapped object for the container
	 * @param payload the object to be wrapped by this <tt>NIOObjectContainer</tt>
	 */
	public NIOObjectContainer(Serializable payload) {
		this.payload = payload;
	}
	



	/**
	 * Serialises the payload into a byte array
	 * @throws IOException If the serialisation process fails
	 */
	private byte[] payloadToByteArray(boolean compressed) throws IOException {
		byte[] payloadBytes;
		Object toWriteOut;
		
		ByteArrayOutputStream os = new ByteArrayOutputStream();
		OutputStream actualStream;
		if(compressed) {
		    actualStream = new GZIPOutputStream(new BufferedOutputStream(os));
		} else {
		    actualStream = os;
		}
		
		ObjectOutputStream helper = new ObjectOutputStream(actualStream);
		
		if( payload != null ) {
			toWriteOut = this.payload;
		} else {
			toWriteOut = EMPTY_PAYLOAD;
		}
		
		try {
			helper.writeObject(toWriteOut);
			helper.flush();
			if(compressed) {
			   ((DeflaterOutputStream)actualStream).finish();
			   ((DeflaterOutputStream)actualStream).flush();
			}
		
			payloadBytes = os.toByteArray();
		} finally {
			helper.close();
			actualStream.close();
			os.close();
		}
		
		return payloadBytes;
		 
	}



	/**
	 * Rebuilds the payload given it serialised form in a byte arrayßß
	 * @param payloadBytes the serialised representation of a payload
	 * @throws IOException if deserialisation  process fails
	 * @throws ClassNotFoundException
	 * @return the deserialised payload
	 */
	private static Serializable byteArrayToPayload(byte[] payloadBytes,boolean compressed) throws IOException,
			ClassNotFoundException {
		ByteArrayInputStream is = new ByteArrayInputStream(payloadBytes);
		InputStream actualIs;
		if(compressed){
		    actualIs = new GZIPInputStream(new BufferedInputStream(is));
		} else {
		    actualIs = is;
		}
		
		ObjectInputStream helper = new ObjectInputStream(actualIs);
		Serializable toReturn = null;
		
		//Testing whether the payload was a null payload.
		try{
			Object payload = helper.readObject();

			if( (payload instanceof Integer) && ((Integer) payload == EMPTY_PAYLOAD)) {
				return null;
			} 
			
			toReturn = (Serializable) payload;

		} finally {
			helper.close();
			actualIs.close();
			is.close();
		}
		
		return toReturn;
	}





	
	
	public byte[] writeContainer(boolean compressed) throws IOException {
	      byte[] encodedHeader = null;
	      byte[] payloadBytes;
	      ByteBuffer message;
	      MessageHeader header;
	      
	      payloadBytes = payloadToByteArray(compressed);
	      header = new MessageHeader(payloadBytes);
	      header.setCompressedPayload(compressed);
	      encodedHeader = header.writeHeader();
	      
	      message = ByteBuffer.allocate(payloadBytes.length+encodedHeader.length);
	      return message.put(encodedHeader).put(payloadBytes).array();

	}
	
	public static NIOObjectContainer readContainer(byte[] encoded) throws IOException, 
	ClassNotFoundException {
	    
	    MessageHeader msgHeader;
	    Serializable payload;
	    ByteBuffer encodedBuffer = ByteBuffer.wrap(encoded);
	    byte[] headerBytes = new byte[MessageHeader.HEADER_SIZE];
	    byte[] payloadBytes;
	    boolean compressed = false;
	    
	    encodedBuffer.get(headerBytes);
	    payloadBytes = new byte[encodedBuffer.remaining()];
	    
	    encodedBuffer.get(payloadBytes);
	    
	    msgHeader = MessageHeader.readHeader(headerBytes);
	    compressed = msgHeader.isCompressedPayload();
	    if(payloadBytes.length != msgHeader.getPayloadLength()) {
		String msg= "Unable to read the correct number of payload bytes";
		log.error(msg + "\n" +
			"It should be: " + msgHeader.getPayloadLength() + " . Read: " + payloadBytes.length);
		throw new StreamCorruptedException(msg);
	    }
	    
	    payload = byteArrayToPayload(payloadBytes,compressed);
	    
	    return new NIOObjectContainer(payload);
	    
	}

	/* (non-Javadoc)
	 * @see java.lang.Object#hashCode()
	 */
	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((payload == null) ? 0 : payload.hashCode());
		return result;
	}

	/* (non-Javadoc)
	 * @see java.lang.Object#equals(java.lang.Object)
	 */
	@Override
	public boolean equals(Object obj) {
	    if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		NIOObjectContainer other = (NIOObjectContainer) obj;
		if (payload == null) {
			if (other.payload != null)
				return false;
		} else if (!payload.equals(other.payload))
			return false;
		return true;
	}
	

	
}


